package com.tvunetworks.center.user.service.impl;

import com.github.pagehelper.PageHelper;
import com.github.pagehelper.PageInfo;
import com.tvunetworks.center.common.constant.LocalConstant;
import com.tvunetworks.center.common.model.ListDeviceParam;
import com.tvunetworks.center.common.model.LoginUser;
import com.tvunetworks.center.common.model.SearchDeviceParam;
import com.tvunetworks.center.common.model.dto.GeoDetail;
import com.tvunetworks.center.common.model.vo.TimeLockDeviceMsgVo;
import com.tvunetworks.center.common.util.*;
import com.tvunetworks.center.gs.model.Device;
import com.tvunetworks.center.user.mapper.DeviceGeoMapper;
import com.tvunetworks.center.user.model.CustomizedUserGroup;
import com.tvunetworks.center.user.model.DeviceGeo;
import com.tvunetworks.center.user.model.param.PageDeviceGeoParam;
import com.tvunetworks.center.user.model.param.SaveDeviceGeoParam;
import com.tvunetworks.center.user.model.vo.DeviceGeoVo;
import com.tvunetworks.center.user.model.vo.PeerIdGroupVo;
import com.tvunetworks.center.user.service.*;
import com.tvunetworks.center.user.service.feign.DeviceFeignService;
import com.tvunetworks.center.user.service.feign.RwsFeignService;
import com.tvunetworks.center.user.util.ThreadPoolUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.text.DecimalFormat;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @program: tvucc-aggregator
 * @description: DeviceGeoServiceImpl
 * @author lebronchen
 * @create: 2019-06-17 18:58
 **/
@Service
@Slf4j
public class DeviceGeoServiceImpl extends BaseService<DeviceGeo> implements DeviceGeoService {

    @Autowired
    private DeviceGeoMapper deviceGeoMapper;
    @Autowired
    private BookmarkDeviceService bookmarkDeviceService;
    @Autowired
    private UserHttpService userHttpService;
    @Autowired
    private DeviceService deviceService;
    @Autowired
    private UserDeviceService userDeviceService;
    @Autowired
    private CustomizedUserDeviceGroupService customizedUserDeviceGroupService;
    @Autowired
    private CustomizedUserGroupService customizedUserGroupService;
    @Autowired
    private RwsFeignService rwsFeignService;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    @Autowired
    private DeviceFeignService deviceFeignService;

    @Value("${enableCustomizedCC}")
    private boolean enableCustomizedCC;
    @Value("${redisKey.receiverCheckUDP}")
    private String receiverCheckUDPKey;

    @Override
    public PageInfo<DeviceGeoVo> pageDeviceGeo(LoginUser loginUser, SearchDeviceParam param, PageDeviceGeoParam deviceGeoParam,
                                               Integer pageNum, Integer pageSize, boolean timeLockFlag) {
        param.setLiving(true);
        param.setOnline(true);
        param.setOffline(true);
        param.setNoCamera(true);
        deviceGeoParam.setUserId(loginUser.getId());
        String role = loginUser.getRole();
        if (StringUtils.equals(LocalConstant.UserRole.SUPPORT, role) || StringUtils.equals(LocalConstant.UserRole.SUPER_ADMIN, role)) {
            // support 不需要查询 group
            PageInfo<DeviceGeoVo> pageInfo = this.supportPage(deviceGeoParam, pageNum, pageSize);
            if (timeLockFlag) {
                if (!CollectionUtils.isEmpty(pageInfo.getList())) {
                    List<DeviceGeoVo> timeLockDeviceGeoVos = this.getTimeLockDeviceGeoVos(pageInfo.getList(), null,loginUser.getId());
                    pageInfo.setList(timeLockDeviceGeoVos);
                }
            }
            return pageInfo;
        } else {
            // admin 和 user 需要查询 group
            return this.userPage(loginUser.getId(), param, deviceGeoParam, pageNum, pageSize, timeLockFlag);
        }
    }

    @Override
    public PageInfo<DeviceGeoVo> supportPage(PageDeviceGeoParam param, int pageNum, int pageSize) {
        String appendSql = param.appendTypeSql();
        param.setAppendSql(appendSql);
        String condition = param.getCondition();
        condition = MyUtil.formatCondition(condition);
        param.setCondition(condition);
        PageHelper.startPage(pageNum, pageSize);
        List<DeviceGeoVo> list = deviceGeoMapper.supportPage(param);

        for (DeviceGeoVo device : list) {
            int count = deviceGeoMapper.isDevicePairWithUserToken(null, device.getPeerId());
            if (count > 0) {
                device.setUpdateType("0");
            } else if ("R".equals(device.getType()) || "T".equals(device.getType())) {
                device.setUpdateType("1");
            } else if ("X".equals(device.getType())) {
                device.setUpdateType("2");
            }
        }
        // 设置 isConnectCC 属性
        this.deviceGeoSetConnectAndUDP(list, param.getMapType());
        return new PageInfo<>(list);
    }

    @Override
    public PageInfo<DeviceGeoVo> userPage(String userId, SearchDeviceParam param, PageDeviceGeoParam deviceGeoParam,
                                          Integer pageNum, Integer pageSize, boolean timeLockFlag) {
        // 过滤bookmark，type，token
        List<String> allPeerIds = userDeviceService.listUserSourceIds(userId, param);
        if (CollectionUtils.isEmpty(allPeerIds)) {
            return new PageInfo<>();
        }

        String condition = param.getCondition();
        condition = MyUtil.formatCondition(condition);
        param.setCondition(condition);
        // 过滤 DeviceGeo 的 status
        List<DeviceGeoVo> list;
        PageInfo<DeviceGeoVo> pageInfo;
        if (timeLockFlag) {
            List<DeviceGeoVo> filterDeviceGeoVos = deviceGeoMapper.userPageDeviceGeoVos(deviceGeoParam);
            List<DeviceGeoVo> deviceGeoVos = this.getTimeLockDeviceGeoVos(filterDeviceGeoVos, allPeerIds,userId);
            PageHelper.startPage(pageNum, pageSize);
            int startIndex = (pageNum - 1) * pageSize;
            int totalSize = deviceGeoVos.size();
            if (startIndex == 0) {
                list = deviceGeoVos.stream().limit(pageSize).collect(Collectors.toList());
            } else {
                list = deviceGeoVos.stream().skip(startIndex).limit(pageSize).collect(Collectors.toList());
            }
            pageInfo = new PageInfo<>(list);
            pageInfo.setTotal(totalSize);
        } else {
            List<String> filterPeerIds = deviceGeoMapper.userPagePeerIds(deviceGeoParam);
            if (CollectionUtils.isEmpty(filterPeerIds)) {
                return new PageInfo<>();
            }
            list = deviceGeoMapper.listDeviceGeoVoByPeerIds(allPeerIds);
            pageInfo = new PageInfo<>(list);
        }
        if (!CollectionUtils.isEmpty(list)) {
            // 提取设备的ID
            List<String> collect = list.stream().map(DeviceGeoVo::getPeerId).collect(Collectors.toList());
            // 查询设备对应的bookmarks
            //Map<peerIds, bookmarks>
            Map<String, List<String>> stringListMap = bookmarkDeviceService.listUserDeviceBookmarks(userId, collect);
            for (DeviceGeoVo device : list) {
                device.setBookmarks(stringListMap.get(device.getPeerId()));

                int count = deviceGeoMapper.isDevicePairWithUserToken(userId, device.getPeerId());
                if (count > 0) {
                    device.setUpdateType("0");
                } else if ("R".equals(device.getType()) || "T".equals(device.getType())) {
                    device.setUpdateType("1");
                } else if ("X".equals(device.getType())) {
                    device.setUpdateType("2");
                }
            }
            // 如果是定制化 CC，查询 Group
            if (enableCustomizedCC) {
                CustomizedUserGroup defaultGroup = customizedUserGroupService.getUserDefaultGroup(userId);
                if (defaultGroup != null) {
                    // 没有默认分组，那么就没有任何分组
                    Map<String, PeerIdGroupVo> peerIdGroupMap = customizedUserDeviceGroupService.mapPeerIdGroup(userId, collect);
                    for (DeviceGeoVo device : list) {
                        PeerIdGroupVo vo = peerIdGroupMap.get(device.getPeerId());
                        if (vo != null) {
                            // 设置 group
                            device.setGroup(vo.getGroup()).setGroupId(vo.getGroupId());
                        } else {
                            // 设置 默认 group
                            device.setGroup(defaultGroup.getGroup()).setGroupId(defaultGroup.getId());
                        }
                    }
                }
            }
        }
        // 设置 isConnectCC 属性
        this.deviceGeoSetConnectAndUDP(list, param.getMapType());
        return pageInfo;
    }

    /**
     * 获取timeLock信息并排序的设备
     * @param filterDeviceGeoVos
     * @param allPeerIds
     * @return
     */
    private List<DeviceGeoVo> getTimeLockDeviceGeoVos(List<DeviceGeoVo> filterDeviceGeoVos, List<String> allPeerIds,String userId) {

        if (CollectionUtils.isEmpty(filterDeviceGeoVos)) {
            return new ArrayList<>();
        }
        List<String> filterPeerIds = filterDeviceGeoVos.stream().map(DeviceGeoVo::getPeerId).map(String::toLowerCase).collect(Collectors.toList());

        Map<String, DeviceGeoVo> deviceGeoVoMap;
        if (!CollectionUtils.isEmpty(allPeerIds)) {
            allPeerIds.retainAll(filterPeerIds);
            if (CollectionUtils.isEmpty(allPeerIds)) {
                return new ArrayList<>();
            }
            deviceGeoVoMap = filterDeviceGeoVos.stream().filter(s -> allPeerIds.contains(s.getPeerId().toLowerCase()))
                    .collect(Collectors.toMap(DeviceGeoVo::getPeerId, Function.identity(), (s1, s2) -> s2));
        }else {
            deviceGeoVoMap = filterDeviceGeoVos.stream().collect(Collectors.toMap(DeviceGeoVo::getPeerId, Function.identity(), (s1, s2) -> s2));
        }
        List<String> packRids = deviceGeoVoMap.values().stream().filter(s -> StringUtils.equalsIgnoreCase(LocalConstant.DeviceType.R, s.getType())
                ||StringUtils.equalsIgnoreCase(LocalConstant.DeviceType.T, s.getType()))
                .map(DeviceGeoVo::getPeerId).collect(Collectors.toList());

        if (!CollectionUtils.isEmpty(packRids)) {
            //获取用户下所有打开timeLock的receiver
            List<TimeLockDeviceMsgVo> timeLockDeviceList;
            if (CollectionUtils.isEmpty(allPeerIds)) {
                timeLockDeviceList = deviceFeignService.getAllReceiverTimeLockMsg();
            } else {
                timeLockDeviceList = deviceFeignService.listOpenTimeLockReceiverByRids(packRids);
            }

            if (!CollectionUtils.isEmpty(timeLockDeviceList)) {
                timeLockDeviceList.forEach(timeLockDeviceMsgVo -> {
                    String peerId = timeLockDeviceMsgVo.getPeerId().toLowerCase();
                    if (StringUtils.isNotEmpty(peerId)) {
                        DeviceGeoVo deviceGeoVo = deviceGeoVoMap.get(peerId);

                        if (deviceGeoVo != null) {
                            deviceGeoVo.setTimeLockMainServer(timeLockDeviceMsgVo.getMainServer() + 1);
                            deviceGeoVo.setTimeLockStatus(timeLockDeviceMsgVo.getStatus());

                        } else {
                            log.error("getTimeLockDeviceGeoVos>>>>>>>>>>deviceGeoVo is empty,peerId:[{}],deviceGeoVoMap:[{}]",
                                    peerId, deviceGeoVoMap);
                        }
                    }

                });
            }
        }

        List<DeviceGeoVo> deviceGeoVos = new ArrayList<>(deviceGeoVoMap.values());
        deviceGeoVos.sort(Comparator.comparing(DeviceGeoVo::getTimeLockMainServer).reversed().thenComparing(DeviceGeoVo::getName));
        return deviceGeoVos;
    }

    private void deviceGeoSetConnectAndUDP(List<DeviceGeoVo> list,String mapType){
        // 设置 isConnectCC 属性
        try {
            if (!CollectionUtils.isEmpty(list)) {
                List<String> peerIds = list.stream().map(item -> item.getPeerId()).collect(Collectors.toList());
                List<String> connectCCRids = rwsFeignService.filterConnectCCR(peerIds);
                if (!CollectionUtils.isEmpty(connectCCRids)) {
                    for (DeviceGeoVo vo : list) {
                        if (connectCCRids.contains(vo.getPeerId())) {
                            vo.setConnectCC(1);
                        }

                    }
                }
            }
        } catch (Exception e) {
            log.error(e.toString(), e);
        }

        for (DeviceGeoVo vo : list) {
            //百度纠偏
            if(StringUtils.isNotEmpty(mapType) && "baidu".equals(mapType)){
                String position = "";
                if("1".equals(vo.getUpdateType())){
                    position = vo.getAutomatic();
                    if(StringUtils.isNotEmpty(position)){
                        String [] geoInfoStr = position.split(",");
                        double[] geoInfo = GPSUtil.gcj02_To_Bd09(Double.valueOf(geoInfoStr[0]),Double.valueOf(geoInfoStr[1]));
                        position = geoInfo[0]+","+geoInfo[1];
                        vo.setAutomatic(position);
                    }
                }else{
                    position = vo.getManual();
                    if(StringUtils.isNotEmpty(position)){
                        String [] geoInfoStr = position.split(",");
                        double[] geoInfo = GPSUtil.gcj02_To_Bd09(Double.valueOf(geoInfoStr[0]),Double.valueOf(geoInfoStr[1]));
                        position = geoInfo[0]+","+geoInfo[1];
                        vo.setManual(position);
                    }
                }

            }
            String type = vo.getType();
            if (!LocalConstant.DeviceType.R.equals(type)) {
                continue;
            }
            String peerId = vo.getPeerId();
            Device device = deviceService.getCurrentDevice(vo.getPeerId());
            //离线
            if(device ==null){
                vo.setIsUdpConnect("NO");
                continue;
            }else if("0".equals(device.getStatus())){
                vo.setIsUdpConnect("NO");
                continue;
            }
            String increment = stringRedisTemplate.opsForValue().get(receiverCheckUDPKey + peerId);
            if (StringUtils.isNotBlank(increment)) {
                int count = Integer.parseInt(increment);
                vo.setTcpReportCount(count);
                if(count>5){
                    vo.setIsUdpConnect("NO");
                }else {
                    vo.setIsUdpConnect("YES");
                }
            }else {
                vo.setIsUdpConnect("YES");
            }

        }
    }

    @Override
    public DeviceGeoVo getDeviceGeo(String peerId) {
        return deviceGeoMapper.getDeviceGeo(peerId);
    }

    /**
     *
     *   通过pid查询DeviceGeo
     * @param peerIds 参数
     * @return List
     */
    @Override
    public List<com.tvunetworks.center.gs.model.DeviceGeo> getDeviceGeoByPeerIds(List<String> peerIds) {
        if (CollectionUtils.isEmpty(peerIds)) {
            return new ArrayList<>();
        }
        List<DeviceGeo> devicePositionByPeerIds = deviceGeoMapper.getDevicePositionByPeerIds(peerIds);
        List<com.tvunetworks.center.gs.model.DeviceGeo> result = new ArrayList<>();
        if (devicePositionByPeerIds== null || devicePositionByPeerIds.size() == 0){
            return result;
        }
        for (DeviceGeo devicePositionByPeerId : devicePositionByPeerIds) {
            com.tvunetworks.center.gs.model.DeviceGeo deviceGeo = new com.tvunetworks.center.gs.model.DeviceGeo();
            try {
                deviceGeo.setId(devicePositionByPeerId.getId());
                deviceGeo.setPeerId(devicePositionByPeerId.getPeerId());
                deviceGeo.setAutomatic(devicePositionByPeerId.getAutomatic());
                deviceGeo.setManual(devicePositionByPeerId.getManual());
                deviceGeo.setStatus(devicePositionByPeerId.getStatus());
                //PropertyUtils.copyProperties(deviceGeo, devicePositionByPeerId);
            } catch (Exception e) {
                log.error(e.toString(), e);
            }
            result.add(deviceGeo);
        }


        return result;
    }

    @Override
    public DeviceGeo getByPeerId(String peerId) {
        DeviceGeo deviceGeo = new DeviceGeo().setPeerId(peerId);
        return super.selectExistOne(deviceGeo);
    }

    @Override
    @Transactional
    public int saveDeviceGeo(SaveDeviceGeoParam param) {
        String peerId = param.getPeerId();
        String manual = param.getManual();

        // 需要更改位置的设备
        List<String> peerIds = new ArrayList<>();
        // R或者X 设置位置同时，更新相关的R和X的位置
        Device device = deviceService.getCurrentDevice(peerId);
        String type = device.getType();
        if (StringUtils.equals(LocalConstant.DeviceType.R, type) || StringUtils.equals(LocalConstant.DeviceType.X, type)) {
            String rid = device.getPeerId().substring(0, 16);
            peerIds = deviceGeoMapper.listRXPeerIdByRid(rid);
        } else {
            peerIds.add(peerId);
        }

        String[] split = manual.split(",");
        if (split.length != 2) {
            // 参数错误
            return 2;
        }
        DecimalFormat df = new DecimalFormat("#.00000");

        String latitude = df.format(Double.parseDouble(split[0]));
        String longitude = df.format(Double.parseDouble(split[1]));
        manual = latitude + "," + longitude;
        for (String pid : peerIds) {
            DeviceGeo deviceGeo = new DeviceGeo();
            deviceGeo.setPeerId(pid);
            DeviceGeo exist = this.getByPeerId(pid);
            if (exist == null) {
                // DeviceGeo 不存在，创建
                deviceGeo.setStatus(2);
                deviceGeo.setManual(manual);
                super.insert(deviceGeo);
            } else {
                // DeviceGeo 已存在，修改
                exist.setManual(manual);
                exist.setStatus(2);
                super.update(exist);
            }
            // X 不发送位置到tpds
            if (pid.length() == 16) {
                userHttpService.sendGeoToTpds(pid, latitude, longitude);
            }
        }
        return 0;
    }

    @Override
    public Result<String> getGeoNameByDevice(String peerId) {
        Device device = deviceService.getCurrentDevice(peerId);
        if (device == null) {
            return ResultUtil.error("", "Device not exist");
        }
        String ip = device.getIp();
        if (StringUtils.isBlank(ip)) {
            return ResultUtil.error("", "Device ip not exist");
        }
        Result<GeoDetail> getGeoNameResult = userHttpService.getGeoNameByIp(ip);
        if (!getGeoNameResult.assertSuccess()) {
            return ResultUtil.error("", "Geo geo name from geo service failed");
        }
        GeoDetail result = getGeoNameResult.getResult();
        String address = null;
        if (result != null) {
            address = result.getAddress();
        }
        return ResultUtil.success(address);
    }

    @Override
    public GeoDetail getGeoByDeviceIp(String peerId) {
        Device device = deviceService.getCurrentDevice(peerId);
        if (device == null) {
            return null;
        }
        String ip = device.getIp();
        if (StringUtils.isBlank(ip)) {
            return null;
        }
        Result<GeoDetail> result = userHttpService.getGeoNameByIp(ip);
        return result.getResult();
    }

    @Override
    public void initDeviceGeo() {

    }

    @Override
    public PageInfo<DeviceGeoVo> pageDeviceGeo(LoginUser loginUser, SearchDeviceParam param, PageDeviceGeoParam deviceGeoParam) {
        //1.分页查询GeoDevice
        PageInfo<DeviceGeoVo> deviceInfo = this.pageDeviceGeo(loginUser, param, deviceGeoParam, param.getPageNum(), param.getPageSize(), Boolean.TRUE);
        List<DeviceGeoVo> deviceGeoVoList = deviceInfo.getList();
        if (CollectionUtils.isEmpty(deviceGeoVoList)) {
            return deviceInfo;
        }
        //2.获取设备的Device对象,Map中的key为peerId,Value为Device
        Map<String, Device> deviceMap = new HashMap<>();
        try {
            String peerIdsString = deviceGeoVoList.stream().map(DeviceGeoVo::getPeerId).collect(Collectors.joining(","));
            ListDeviceParam listDeviceParam = new ListDeviceParam();
            listDeviceParam.setPeerIds(peerIdsString);
            List<Device> devices = deviceFeignService.listCurrentDevice(listDeviceParam);
            if (!CollectionUtils.isEmpty(devices)) {
                deviceMap = devices.stream().collect(Collectors.toMap(Device::getPeerId, Function.identity(), (s1, s2) -> s2));
            }
        } catch (Exception e) {
            log.error("pageDeviceGeo>>>>>>>>>>getDeviceListException,param:[{}],e:[{}]", param, e);
        }
        //3.获取设备的地理位置
        this.getDeviceGeoAndLastReportTime(deviceGeoVoList, deviceMap, param.isAddress());
        return deviceInfo;
    }

    /**
     * 根据deviceGeo对象获取该设备的地理位置和最后一次上报时间
     *
     * @param deviceGeoVoList deviceGeo
     * @param deviceMap       device设备的对应关系
     * @param address         是否需要获取地址
     */
    private void getDeviceGeoAndLastReportTime(List<DeviceGeoVo> deviceGeoVoList, Map<String, Device> deviceMap, boolean address) {
        int size = deviceGeoVoList.size();
        for (int i = 0; i < size / 10 + 1; i++) {
            int indexStart = 10 * i;
            List<DeviceGeoVo> deviceGeoVos;
            if (indexStart == 0) {
                deviceGeoVos = deviceGeoVoList.stream().limit(10).collect(Collectors.toList());
            }else {
                deviceGeoVos = deviceGeoVoList.stream().skip(indexStart).limit(10).collect(Collectors.toList());
            }
            final CountDownLatch countDownLatch = new CountDownLatch(deviceGeoVos.size());
            deviceGeoVos.forEach(deviceGeoVo -> ThreadPoolUtil.execute(()->{
                try {
                    Device device = deviceMap.get(deviceGeoVo.getPeerId());
                    if (device != null) {
                        deviceGeoVo.setVersion(device.getVersion());
                        deviceGeoVo.setDeviceStatus(device.getStatus());
                    }
                    if (address) {
                        try {
                            Result<String> name = null;
                            if (!StringUtils.equalsIgnoreCase("0,0", deviceGeoVo.getManual()) && StringUtils.isNotEmpty(deviceGeoVo.getManual())) {
                                name = userHttpService.getGeoNameByLocation(deviceGeoVo.getManual());
                            } else if (!StringUtils.equalsIgnoreCase("0,0", deviceGeoVo.getAutomatic()) && StringUtils.isNotEmpty(deviceGeoVo.getAutomatic())) {
                                name = userHttpService.getGeoNameByLocation(deviceGeoVo.getAutomatic());
                            }
                            if (name != null) {
                                deviceGeoVo.setAddress(name.getResult());
                            }
                            try {
                                Result<String> result = deviceFeignService.listGPSByMultiPeerId(deviceGeoVo.getPeerId());
                                if (result != null && result.assertSuccess() && StringUtils.isNotEmpty(result.getResult())) {
                                    String result1 = result.getResult();
                                    List<Map> listMap = JsonUtil.toObject(result1, List.class);
                                    if (!CollectionUtils.isEmpty(listMap)) {
                                        Map map = listMap.get(0);
                                        if (map.get("timestamp") != null) {
                                            deviceGeoVo.setGeoReportTime(Long.parseLong(map.get("timestamp").toString()));
                                        }
                                    }
                                } else {
                                    log.error("pageDeviceGeo>>>>getGeoTimestamp11>>>>>> failed,peerId:[{}],result:[{}]", deviceGeoVo.getPeerId(), result);
                                }
                            } catch (Exception e) {
                                log.error("pageDeviceGeo>>>>>>getGeoTimestamp11>>>> peerId:[{}],exception:[{}]", deviceGeoVo.getPeerId(), e);
                            }
                        } catch (Exception e) {
                            log.error("==userHttpService.getGeoNameByLocation==", e);
                        }
                    }
                } finally {
                    countDownLatch.countDown();
                }
            }));
            try {
                countDownLatch.await();
            }catch (Exception e){
                log.error("getDeviceGeoAndLastReportTime>>>>>>countLatch Exception", e);
                countDownLatch.countDown();
            }
        }
    }

}
